import json
from time import time

from kafka import KafkaConsumer


class KConsumer:
    def __init__(self, bootstrap_servers, topic):
        """
        kafka 生产者
        :param bootstrap_servers: 地址
        :param topic:  topic
        """
        self.consumer = KafkaConsumer(topic, bootstrap_servers=bootstrap_servers, auto_offset_reset='latest',
                                      enable_auto_commit=False)  # json 格式化发送的内容
        self.topic = topic

    def sync_consumer(self):
        """
        同步发送 数据
        :param data_li:  发送数据
        :return:
        """
        # self.consumer.seek_to_end()
        for message in self.consumer:
            print("message", self.consumer)
            return message

    def send_success(self, *args, **kwargs):
        """异步发送成功回调函数"""
        print('save success')
        return

    def send_error(self, *args, **kwargs):
        """异步发送错误回调函数"""
        print('save error')
        return

    def close_producer(self):
        try:
            self.consumer.close()
        except:
            pass


if __name__ == '__main__':
    send_data_li = [{"test": 1}, {"test": 2}]
    kp = KConsumer(topic='51_receive_tess', bootstrap_servers='10.102.1.113:9092')

    # 同步接收
    kp.sync_consumer()
